package rabbitmq.confirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.sun.source.tree.SynchronizedTree;
import com.sun.source.tree.Tree;
import rabbitmq.constants.Constants;
import rabbitmq.utils.WorkUtils;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

public class ProducerConfirm {
    public static final Integer MESSAGE_SIZE = 200;
    public static final Integer BATCH_SIZE = 100;


    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        //Strategy #1: Publishing Messages Individually
        //单独确认
        publishingMessagesIndividually();
        //Strategy #2: Publishing Messages in Batches
        //批量确认
        publishingMessagesInBatches();
        //Strategy #3: Handling Publisher Confirms Asynchronously
        //异步确认
        handlingPublisherConfirmsAsynchronously();
    }

    private static void handlingPublisherConfirmsAsynchronously() throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory connectionFactory = WorkUtils.createConnectionFactory();
        //建立连接
        Connection connection = connectionFactory.newConnection();
        //开启信道
        Channel channel = connection.createChannel();
        //设置信道为confirm模式
        channel.confirmSelect();
        //声明队列
        channel.queueDeclare(Constants.PRODUCER_CONFIRM_QUEUE3,true,false,false,null);
        //监听
        SortedSet<Long> set = Collections.synchronizedSortedSet(new TreeSet<>());
        long start = System.currentTimeMillis();
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                if (multiple){
                    set.headSet(deliveryTag+1).clear();
                }else {
                    set.remove(deliveryTag);
                }
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                if (multiple){
                    set.headSet(deliveryTag+1).clear();
                }else {
                    set.remove(deliveryTag);
                }
            }
        });

        //发送消息
        for (int i = 0; i < MESSAGE_SIZE; i++){
            String msg = "hello handlingPublisherConfirmsAsynchronously~" + i;
            long seqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("",Constants.PRODUCER_CONFIRM_QUEUE3,null,msg.getBytes());
            set.add(seqNo);
        }

        while (!set.isEmpty()){
            Thread.sleep(10);
        }

        long end = System.currentTimeMillis();
        System.out.printf("异步确认模式，消息个数：%d, 耗时：%d \n", MESSAGE_SIZE, end-start);
    }

    private static void publishingMessagesInBatches() throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory connectionFactory = WorkUtils.createConnectionFactory();
        //建立连接
        Connection connection = connectionFactory.newConnection();
        //开启信道
        Channel channel = connection.createChannel();
        //设置信道为confirm模式
        channel.confirmSelect();
        //声明队列
        channel.queueDeclare(Constants.PRODUCER_CONFIRM_QUEUE2,true,false,false,null);
        //发布消息并等待确认
        int count = 0;
        long start = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_SIZE; i++){
            String msg = "hello publishingMessagesIndividually ~" + i;
            channel.basicPublish("",Constants.PRODUCER_CONFIRM_QUEUE2,null,msg.getBytes());
            count++;
            if (count == BATCH_SIZE){
                channel.waitForConfirms(5000);
                count = 0;
            }
        }
        if (count > 0){
            channel.waitForConfirms(5000);
        }
        long end = System.currentTimeMillis();
        System.out.printf("批量确认模式，消息个数：%d, 耗时：%d \n", MESSAGE_SIZE, end-start);
    }

    private static void publishingMessagesIndividually() throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory connectionFactory = WorkUtils.createConnectionFactory();
        //建立连接
        Connection connection = connectionFactory.newConnection();
        //开启信道
        Channel channel = connection.createChannel();
        //设置信道为confirm模式
        channel.confirmSelect();
        //声明队列
        channel.queueDeclare(Constants.PRODUCER_CONFIRM_QUEUE1,true,false,false,null);
        //发布消息并等待确认
        long start = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_SIZE; i++){
            String msg = "hello publishingMessagesIndividually ~" + i;
            channel.basicPublish("",Constants.PRODUCER_CONFIRM_QUEUE1,null,msg.getBytes());
            channel.waitForConfirms(5000);
        }
        long end = System.currentTimeMillis();
        System.out.printf("单独确认模式，消息个数：%d, 耗时：%d \n", MESSAGE_SIZE, end-start);
    }
}
